/*
Copyright 2015 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
	"errors"
	"fmt"
	"log"
	"net/http"
	"tmlake.com/api-gateway/istio-agent/pkg/model"
	"tmlake.com/api-gateway/istio-agent/proxy/file"
	"tmlake.com/api-gateway/istio-agent/proxy/njetcontroller/configs"
	njt_template "tmlake.com/api-gateway/istio-agent/proxy/njetcontroller/controller/template"
	"tmlake.com/api-gateway/istio-agent/xds/xdsclient/xdsresource"

	"net"
	"os"
	"os/exec"
	"sync"
	"syscall"
	"time"

	"github.com/eapache/channels"
	njt_log "tmlake.com/api-gateway/istio-agent/proxy"
	"tmlake.com/api-gateway/istio-agent/proxy/njet"
	njt_config "tmlake.com/api-gateway/istio-agent/proxy/njetcontroller/controller/config"
	"tmlake.com/api-gateway/istio-agent/proxy/njetcontroller/controller/process"

	"tmlake.com/api-gateway/istio-agent/xds/resolver"
)

const (
	tempNginxPattern = "njet-cfg"
	emptyUID         = "-1"
)

// NewNJETController creates a new NJET controller.
//func NewNJETController(config *Configuration) *NJETController {
func NewNJETController(config *Configuration, node *model.Node) *NJETController {
	n := &NJETController{
		cfg:           config,
		stopCh:        make(chan struct{}),
		updateCh:      channels.NewRingChannel(1024),
		ngxErrCh:      make(chan error),
		stopLock:      &sync.Mutex{},
		command:       NewNginxCommand(),
		backendConfig: njt_config.NewDefault(),
		runningConfig: new(configs.Configuration),
	}

	n.cfg.Node = node
	r, err := resolver.New(n.updateCh, node)
	if err != nil {
		njt_log.NjetLogger.Errorf("New resolver failed: %v", err)
		return nil
	}

	var ok bool
	n.r, ok = r.(*resolver.XdsResolver)
	if !ok {
		njt_log.NjetLogger.Fatalf("expected a '*resolver.XdsResolver' type but %T was returned", r)
	}
	ngxTpl, err := njt_template.NewTemplate(njet.TemplatePath)
	if err != nil {
		njt_log.NjetLogger.Fatalf("Invalid NJET configuration template: %v", err)
	}

	n.t = ngxTpl

	return n
}

// NJETController describes a NJET controller.
type NJETController struct {
	backendConfig njt_config.Configuration
	r             *resolver.XdsResolver
	cfg           *Configuration

	// stopLock is used to enforce that only a single call to Stop send at
	// a given time. We allow stopping through an HTTP endpoint and
	// allowing concurrent stoppers leads to stack traces.
	stopLock *sync.Mutex

	stopCh   chan struct{}
	updateCh *channels.RingChannel

	// ngxErrCh is used to detect errors with the NJET processes
	ngxErrCh chan error

	// runningConfig contains the running configuration in the Backend
	runningConfig *configs.Configuration

	virtualInboundListenerConfig *xdsresource.InboundListenerConfig

	t njt_template.Writer

	resolver []net.IP

	isIPV6Enabled bool

	isShuttingDown bool

	command NginxExecTester
}

// Start starts a new NJET master process running in the foreground.
func (n *NJETController) Start() {
	njt_log.NjetLogger.Infof("Starting NJET controller...")

	cmd := n.command.ExecCommand()

	if cmd == nil {
		njt_log.NjetLogger.Fatalf("Exec njet cmd nil")
	}

	// put NJET in another process group to prevent it
	// to receive signals meant for the controller
	cmd.SysProcAttr = &syscall.SysProcAttr{
		Setpgid: true,
		Pgid:    0,
	}

	n.start(cmd)

	for {
		select {
		case err := <-n.ngxErrCh:
			if n.isShuttingDown {
				return
			}

			// if the njet master process dies, the workers continue to process requests
			// until the failure of the configured livenessProbe and restart of the pod.
			if process.IsRespawnIfRequired(err) {
				njt_log.NjetLogger.Fatalf("NJET error: %v", err)
				return
			}

		case event := <-n.updateCh.Out():
			if n.isShuttingDown {
				break
			}

			njt_log.NjetLogger.Infof("start syncXDS, event:%+v", event)

			n.syncXDS(event.(resolver.Event).Obj)
		case <-n.stopCh:
			return
		}
	}
}

// Stop gracefully stops the NJET master process.
func (n *NJETController) Stop() error {
	n.isShuttingDown = true

	n.stopLock.Lock()
	defer n.stopLock.Unlock()

	time.Sleep(time.Duration(n.cfg.ShutdownGracePeriod) * time.Second)

	close(n.stopCh)

	// send stop signal to NJET
	njt_log.NjetLogger.Infof("Stopping NJET process")
	log.Println("Stopping NJET process")
	cmd := n.command.ExecCommand("-s", "quit")
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	err := cmd.Run()
	if err != nil {
		return err
	}

	// wait for the NJET process to terminate
	timer := time.NewTicker(time.Second * 1)
	for range timer.C {
		if !njet.IsRunning() {
			njt_log.NjetLogger.Infof("NJET process has stopped")
			timer.Stop()
			break
		}
	}

	return nil
}

func (n *NJETController) start(cmd *exec.Cmd) {
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		n.ngxErrCh <- err
		njt_log.NjetLogger.Fatalf("lllNJET error: %v", err)
		return
	}
	go func() {
		n.ngxErrCh <- cmd.Wait()
	}()
}

// generateTemplate returns the njet configuration file content
func (n NJETController) generateTemplate(cfg njt_config.Configuration, xdsCfg configs.Configuration) ([]byte, error) {

	// NJET cannot resize the hash tables used to store server names. For
	// this reason we check if the current size is correct for the host
	// names defined in the Ingress rules and adjust the value if
	// necessary.
	// https://trac.njet.org/njet/ticket/352
	// https://trac.njet.org/njet/ticket/631
	var longestName int
	var serverNameBytes int
	var hostnameLength int

	for _, srvCfg := range xdsCfg.Servers {
		for _, hostname := range srvCfg.Server.Hostnames {
			hostnameLength := len(hostname)
			if srvCfg.Server.RedirectFromToWWW {
				hostnameLength += 4
			}
			if longestName < hostnameLength {
				longestName = hostnameLength
			}
		}

		for _, alias := range srvCfg.Server.Aliases {
			if longestName < len(alias) {
				longestName = len(alias)
			}
		}

		serverNameBytes += hostnameLength
	}

	nameHashBucketSize := njetHashBucketSize(longestName)
	if cfg.ServerNameHashBucketSize < nameHashBucketSize {
		njt_log.NjetLogger.Info("Adjusting ServerNameHashBucketSize variable", "value", nameHashBucketSize)
		cfg.ServerNameHashBucketSize = nameHashBucketSize
	}

	serverNameHashMaxSize := nextPowerOf2(serverNameBytes)
	if cfg.ServerNameHashMaxSize < serverNameHashMaxSize {
		njt_log.NjetLogger.Info("Adjusting ServerNameHashMaxSize variable", "value", serverNameHashMaxSize)
		cfg.ServerNameHashMaxSize = serverNameHashMaxSize
	}

	if cfg.MaxWorkerOpenFiles == 0 {
		// the limit of open files is per worker process
		// and we leave some room to avoid consuming all the FDs available
		maxOpenFiles := rlimitMaxNumFiles() - 1024
		njt_log.NjetLogger.Info("Maximum number of open file descriptors", "value", maxOpenFiles)
		if maxOpenFiles < 1024 {
			// this means the value of RLIMIT_NOFILE is too low.
			maxOpenFiles = 1024
		}
		njt_log.NjetLogger.Info("Adjusting MaxWorkerOpenFiles variable", "value", maxOpenFiles)
		cfg.MaxWorkerOpenFiles = maxOpenFiles
	}

	if cfg.MaxWorkerConnections == 0 {
		maxWorkerConnections := int(float64(cfg.MaxWorkerOpenFiles * 3.0 / 4))
		njt_log.NjetLogger.Info("Adjusting MaxWorkerConnections variable", "value", maxWorkerConnections)
		cfg.MaxWorkerConnections = maxWorkerConnections
	}

	setHeaders := map[string]string{}
	addHeaders := map[string]string{}

	tc := njt_config.TemplateConfig{
		ProxySetHeaders:          setHeaders,
		AddHeaders:               addHeaders,
		BacklogSize:              sysctlSomaxconn(),
		Backends:                 xdsCfg.Backends,
		Servers:                  xdsCfg.Servers,
		TCPL4Services:            xdsCfg.TCPL4Services,
		UDPL4Services:            xdsCfg.UDPL4Services,
		Cfg:                      cfg,
		IsIPV6Enabled:            n.isIPV6Enabled && !cfg.DisableIpv6,
		NginxStatusIpv4Whitelist: cfg.NginxStatusIpv4Whitelist,
		NginxStatusIpv6Whitelist: cfg.NginxStatusIpv6Whitelist,
		ListenPorts:              n.cfg.ListenPorts,
		HealthzURI:               njet.HealthPath,
		PID:                      njet.PID,
		StatusPath:               njet.StatusPath,
		StatusPort:               njet.StatusPort,
		StreamPort:               njet.StreamPort,
	}

	tc.Cfg.Checksum = xdsCfg.ConfigurationChecksum

	return n.t.Write(tc)
}

// testTemplate checks if the NJET configuration inside the byte array is valid
// running the command "njet -t" using a temporal file.
func (n NJETController) testTemplate(cfg []byte) error {
	if len(cfg) == 0 {
		return fmt.Errorf("invalid NJET configuration (empty)")
	}
	tmpDir := os.TempDir() + "/njet"
	tmpfile, err := os.CreateTemp(tmpDir, tempNginxPattern)
	if err != nil {
		return err
	}
	defer tmpfile.Close()
	err = os.WriteFile(tmpfile.Name(), cfg, file.ReadWriteByUser)
	if err != nil {
		return err
	}
	out, err := n.command.Test(tmpfile.Name())
	if err != nil {
		// this error is different from the rest because it must be clear why njet is not working
		oe := fmt.Sprintf(`
-------------------------------------------------------------------------------
Error: %v
%v
-------------------------------------------------------------------------------
`, err, string(out))

		return errors.New(oe)
	}

	os.Remove(tmpfile.Name())
	return nil
}

// njetHashBucketSize computes the correct NJET hash_bucket_size for a hash
// with the given longest key.
func njetHashBucketSize(longestString int) int {
	// see https://github.com/kubernetes/configs-njets/issues/623 for an explanation
	wordSize := 8 // Assume 64 bit CPU
	n := longestString + 2
	aligned := (n + wordSize - 1) & ^(wordSize - 1)
	rawSize := wordSize + wordSize + aligned
	return nextPowerOf2(rawSize)
}

// http://graphics.stanford.edu/~seander/bithacks.html#RoundUpPowerOf2
// https://play.golang.org/p/TVSyCcdxUh
func nextPowerOf2(v int) int {
	v--
	v |= v >> 1
	v |= v >> 2
	v |= v >> 4
	v |= v >> 8
	v |= v >> 16
	v++

	return v
}

// Helper function to clear Certificates from the configs configuration since they should be ignored when
// checking if the new configuration changes can be applied dynamically if dynamic certificates is on
func clearCertificates(config *configs.Configuration) {
	var clearedServers []*configs.VirtualServerConfig
	for _, server := range config.Servers {
		copyOfServer := *server
		copyOfServer.Server.SSLCert = nil
		clearedServers = append(clearedServers, &copyOfServer)
	}
	config.Servers = clearedServers
}

func clearBackendsExceptTLSMode(config *configs.Configuration) {
	var clearedBackends []*configs.Backend
	for _, b := range config.Backends {
		copyofBackend := &configs.Backend{
			Name:    b.Name,
			SSLMode: b.SSLMode,
		}
		clearedBackends = append(clearedBackends, copyofBackend)
	}

	config.Backends = clearedBackends
}

// Helper function to clear endpoints from the configs configuration since they should be ignored when
// checking if the new configuration changes can be applied dynamically.
func clearL4serviceEndpoints(config *configs.Configuration) {
	var clearedTCPL4Services = make(map[string]*configs.L4Service)
	var clearedUDPL4Services []configs.L4Service
	for listenerName, service := range config.TCPL4Services {
		copyofService := &configs.L4Service{
			Port:                                  service.Port,
			Endpoints:                             []configs.Endpoint{},
			SSLCert:                               service.SSLCert,
			CertificateAuth:                       service.CertificateAuth,
			DestinationPortPeerAuthenticationMode: service.DestinationPortPeerAuthenticationMode,
		}
		clearedTCPL4Services[listenerName] = copyofService
	}
	for _, service := range config.UDPL4Services {
		copyofService := configs.L4Service{
			Port:      service.Port,
			Endpoints: []configs.Endpoint{},
		}
		clearedUDPL4Services = append(clearedUDPL4Services, copyofService)
	}
	config.TCPL4Services = clearedTCPL4Services
	config.UDPL4Services = clearedUDPL4Services
}

// IsDynamicConfigurationEnough returns whether a Configuration can be
// dynamically applied, without reloading the backend.
func (n *NJETController) IsDynamicConfigurationEnough(pcfg *configs.Configuration) bool {
	copyOfRunningConfig := *n.runningConfig
	copyOfPcfg := *pcfg

	//copyOfRunningConfig.Backends = []*configs.Backend{}
	//copyOfPcfg.Backends = []*configs.Backend{}

	//The clientTLSMode cannot be dynamic, But because the back length is not equal, rather than the tlsmode change, it causes reload
	clearBackendsExceptTLSMode(&copyOfRunningConfig)
	clearBackendsExceptTLSMode(&copyOfPcfg)

	clearL4serviceEndpoints(&copyOfRunningConfig)
	clearL4serviceEndpoints(&copyOfPcfg)

	//The certificate cannot be dynamic
	//clearCertificates(&copyOfRunningConfig)
	//clearCertificates(&copyOfPcfg)

	return copyOfRunningConfig.Equal(&copyOfPcfg)
}

// configureDynamically encodes new Backends in JSON format and POSTs the
// payload to an internal HTTP endpoint handled by Lua.
func (n *NJETController) configureDynamically(pcfg *configs.Configuration) error {
	err := configureBackends(pcfg.Backends)
	return err
}

func configureBackends(rawBackends []*configs.Backend) error {
	backends := make([]*configs.Backend, len(rawBackends))

	for i, backend := range rawBackends {
		luaBackend := &configs.Backend{
			Name:                 backend.Name,
			SSLPassthrough:       backend.SSLPassthrough,
			SessionAffinity:      backend.SessionAffinity,
			LoadBalancing:        backend.LoadBalancing,
			NoServer:             backend.NoServer,
			TrafficShapingPolicy: backend.TrafficShapingPolicy,
			AlternativeBackends:  backend.AlternativeBackends,
		}

		var endpoints []configs.Endpoint
		for _, endpoint := range backend.Endpoints {
			endpoints = append(endpoints, configs.Endpoint{
				Address: endpoint.Address,
				Port:    endpoint.Port,
			})
		}

		luaBackend.Endpoints = endpoints
		backends[i] = luaBackend
	}

	statusCode, _, err := njet.NewPostStatusRequest("/configuration/backends", "application/json", backends)
	if err != nil {
		return err
	}

	if statusCode != http.StatusCreated {
		return fmt.Errorf("unexpected error code: %d", statusCode)
	}

	return nil
}
